package com.sf.iisp.kafka.producer;

import java.util.List;

import com.sf.kafka.api.produce.IKafkaProducer;
import com.sf.kafka.api.produce.ProduceConfig;
import com.sf.kafka.api.produce.ProducerPool;
import com.sf.novatar.context.AppContext;

/**
 * Created by qi on 16-7-6.
 */
@Deprecated
public class SGSProducer {
    // 生产者连接池大小
    private static final int poolSize = Integer.valueOf(AppContext.getContext().getConfigProperties().getProperty("kafka.messageConsumeThreadCount", "1"));
    // 主题名称
    private static final String topic = AppContext.getContext().getConfigProperties().getProperty("kafka.topic", "ESG_IISP_SGS");
    // KAFKA连接地址
    private static final String url = AppContext.getContext().getConfigProperties().getProperty("kafka.url", "http://10.202.34.30:8292/mom-mon/monitor/requestService.pub");
    // 主题所在的集群名称
    private static final String clusterName = AppContext.getContext().getConfigProperties().getProperty("kafka.clusterName", "other");
    // 主题名称＋分隔符 （固定不变）＋主题的校验码
    private static final String topicTokens = topic + ":" + AppContext.getContext().getConfigProperties().getProperty("kafka.topicTokens", "Q9E2*smR");


    public static void send(List<String> messages) {
        if (messages == null || messages.size() == 0) {
            return;
        }
        //初始化KAFKA内容
        ProduceConfig produceConfig = new ProduceConfig(poolSize, url, clusterName, topicTokens);
        IKafkaProducer kafkaProducer = new ProducerPool(produceConfig);
        //发送KAFKA消息
        kafkaProducer.batchSendString(topic, messages);
        kafkaProducer.close();
    }


}
